package com.wenge.datagroup.storage.util;

import com.wenge.datagroup.storage.dataprocess.MultilingualAbstract;
import com.wenge.datagroup.storage.dataprocess.MultilingualEmotion;
import com.wenge.datagroup.storage.dataprocess.MultilingualEntity;
import com.wenge.datagroup.storage.dataprocess.MultilingualKeywords;
import com.wenge.datagroup.storage.dataprocess.MultilingualSpiltWords;

import org.apache.kafka.common.TopicPartition;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStreamReader;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Properties;
import java.util.Set;


public class ConfigUtil {

    public static Properties properties = new Properties();
    static String configFile = "config" + File.separator + "config.properties";
    private static String topic;

    public static void setTopic(String topic) {
        ConfigUtil.topic = topic;
    }

    public static void setConfigFile(String configFilePath) {
        configFile = configFilePath;
    }

    public static Set<TopicPartition> topic_partitions_set = new HashSet<>();
    public static Set<String> topic_partitionSet = new HashSet<>();

    public static Properties sparkConf = new Properties();
    public static String nodeIp = "";
    public static int sendRepartition = 3;

    /* Kafka/Zookeeper 相关配置  */
    public static String bootstrapServers;

    public static String languageUrl;
    public static String translationUrl;
    static {
        if (nodeIp == null || nodeIp.isEmpty()) {
            try {
                nodeIp = IpUtil.getHostIp();
            } catch (Exception e1) {
                e1.printStackTrace();
            }
        }
    }

    public static Map<String, String> getSparkConfig(String filePath) {
        Map<String, String> confMap = new HashMap<>();
        if (filePath != null && !filePath.isEmpty()) {
            FileInputStream fileStream = null;
            try {
                fileStream = new FileInputStream(filePath);
                sparkConf.load(new InputStreamReader(fileStream, "UTF-8"));
                for (Object k : sparkConf.keySet()) {
                    String v = sparkConf.getProperty(k.toString());
                    confMap.put(k.toString(), v);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        return confMap;
    }

    public static void initTopicPartitions() {
        if (!topic_partitionSet.isEmpty()) {
            System.out.println("topic_partitionSet=" + topic_partitionSet);
            for (String p : topic_partitionSet) {
                topic_partitions_set.add(new TopicPartition(ConfigUtil.topic, Integer.valueOf(p)));
            }
        }
    }

    public static void initTopicPartitionSet(String partitions) {
        if (partitions != null) {
            for (String p : partitions.split(",")) {
                topic_partitionSet.add(p);
            }
        }
    }

    public static void init() {

        try {
            FileInputStream fileStream = new FileInputStream(configFile);
            properties.load(new InputStreamReader(fileStream, "UTF-8"));

            //Kafka/Zookeeper 相关配置
            bootstrapServers = properties.getProperty("bootstrapServers").trim();
            String generalUrl = properties.get("generalUrl").toString().trim();
            String anAbstract = properties.get("abstract").toString().trim();
            String emotion = properties.get("emotion").toString().trim();
            String entity = properties.get("entity").toString().trim();
            String keyword = properties.get("keyword").toString().trim();
            String language = properties.get("language").toString().trim();
            String splitWord = properties.get("splitWord").toString().trim();
            String translation = properties.get("translation").toString().trim();
            MultilingualAbstract.setUrl(generalUrl + anAbstract);
            MultilingualEmotion.setUrl(generalUrl + emotion);
            MultilingualEntity.setUrl11(generalUrl + entity);
            MultilingualKeywords.setUrl(generalUrl + keyword);
            languageUrl = language;
            MultilingualSpiltWords.setUrl(generalUrl + splitWord);
            translationUrl = translation;
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(0);
        }
        initTopicPartitions();
    }


    public static String get(String key) {
        return properties.getProperty(key);
    }
}
